package cn.doitedu.demo.stream_table;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.planner.typeutils.RowTypeUtils;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.types.RowUtils;
import org.apache.flink.util.Collector;

import java.util.LinkedHashMap;

public class Changelog流转表 {
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .databaseList("flinktest") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("flinktest.flink_score") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();


        DataStreamSource<String> stream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "cdc");


        // 从json中解析出数据，封装成flinkSql流转表所需要Row类型
        //Row tmpRow = Row.ofKind(RowKind.UPDATE_BEFORE, 1, "zs", "male", 80.2);
        //TypeInformation<Row> typeInfo = TypeExtractor.getForObject(tmpRow);


        TypeInformation[] typeInformations = new TypeInformation[4];
        typeInformations[0] = TypeInformation.of(Integer.class); // id 是整数
        typeInformations[1] = TypeInformation.of(String.class); // id 是整数
        typeInformations[2] = TypeInformation.of(String.class); // id 是整数
        typeInformations[3] = TypeInformation.of(Double.class); // id 是整数
        RowTypeInfo typeInfo = new RowTypeInfo(typeInformations);


        SingleOutputStreamOperator<Row> rowStream = stream.process(new ProcessFunction<String, Row>() {
            @Override
            public void processElement(String cdcJson, ProcessFunction<String, Row>.Context context, Collector<Row> collector) throws Exception {

                JSONObject before;
                JSONObject after;

                JSONObject jsonObject = JSON.parseObject(cdcJson);

                String op = jsonObject.getString("op");

                if (op.equals("u")) {
                    before = jsonObject.getJSONObject("before");
                    int id = before.getInteger("id");
                    String name = before.getString("name");
                    String gender = before.getString("gender");
                    double score = before.getDouble("score");
                    // 生成一条rowKind= -U 的row
                    Row updateBeforeRow = Row.ofKind(RowKind.UPDATE_BEFORE, id, name, gender, score);
                    collector.collect(updateBeforeRow);


                    after = jsonObject.getJSONObject("after");
                    id = after.getInteger("id");
                    name = after.getString("name");
                    gender = after.getString("gender");
                    score = after.getDouble("score");
                    // 生成一条rowKind= +U 的row
                    Row updateAfterRow = Row.ofKind(RowKind.UPDATE_AFTER, id, name, gender, score);
                    collector.collect(updateAfterRow);

                } else if (op.equals("c") || op.equals("r")) {

                    after = jsonObject.getJSONObject("after");

                    int id = after.getInteger("id");
                    String name = after.getString("name");
                    String gender = after.getString("gender");
                    double score = after.getDouble("score");
                    // 生成一条rowKind= +U 的row
                    Row insertRow = Row.ofKind(RowKind.INSERT, id, name, gender, score);
                    collector.collect(insertRow);
                } else {


                    before = jsonObject.getJSONObject("before");
                    int id = before.getInteger("id");
                    String name = before.getString("name");
                    String gender = before.getString("gender");
                    double score = before.getDouble("score");
                    // 生成一条rowKind= -D 的row
                    Row deleteRow = Row.ofKind(RowKind.DELETE, id, name, gender, score);

                    collector.collect(deleteRow);
                }
            }
        }).returns(typeInfo);


        // 将上面的row类型的流，转成 表
        Table table = tenv.fromChangelogStream(rowStream);
        tenv.createTemporaryView("xx",table);

        tenv.executeSql("select f0 as id,f1 as name,f2 as gender ,f3 as score from xx").print();


        env.execute();
        System.exit(1);



        // 处理上面的流：输出所有binlog日志中的 after里面的数据内容，如果after==null，则输出before的内容
        // op,1,taoge,male,999
        SingleOutputStreamOperator<String> processed = stream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String json, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {

                JSONObject jsonObject = JSON.parseObject(json);
                // 取操作类型
                String op = jsonObject.getString("op");

                // 取表数据
                JSONObject dataJsonObject;
                if (op.equals("d")) {
                    dataJsonObject = jsonObject.getJSONObject("before");
                } else {
                    dataJsonObject = jsonObject.getJSONObject("after");
                }


                Integer id = dataJsonObject.getInteger("id");
                String name = dataJsonObject.getString("name");
                String gender = dataJsonObject.getString("gender");
                Double score = dataJsonObject.getDouble("score");


                // 拼接结果返回
                collector.collect(op + "," + id + "," + name + "," + gender + "," + score);
            }
        });

        processed.print();

        env.execute();

    }
}
